package com.atbeijing.bigdata.spark.mytest.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 转换结构(品类,(点击数,下单数,支付数))
 * 利用tuple的排序特点
 */
object Top10_1 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("top")
    val sc = new SparkContext(conf)

    //先按照点击数排名，靠前的就排名高；如果点击数相同，再比较下单数；下单数再相同，就比较支付数。
    val file: RDD[String] = sc.textFile("data/user_visit_action.txt")
    //(品类id,(点击量,0,0))
    //(品类id,(0,下单量,0))
    //(品类id,(0,0,支付量)) =>reduceByKey =>(品类,(点击数,下单数,支付数))
    //一行一行的数据流过flatMap这个算子后格式发生变化
    val r1: RDD[(String, (Int, Int, Int))] = file.flatMap(line => {
      val data: Array[String] = line.split("_")
      if (data(6) != "-1") {
        //点击数据
        List((data(6), (1, 0, 0)))
      } else if (data(8) != "null") {
        // 下单数据
        val ids: Array[String] = data(8).split(",")
        ids.map(
          id => (id, (0, 1, 0))
        )
      } else if (data(10) != "null") {
        //支付数据
        val ids: Array[String] = data(10).split(",")
        ids.map(
          id => (id, (0, 0, 1))
        )
      } else {
        Nil
      }
    })
    r1.reduceByKey(
      (v1,v2)=>{
        (v1._1+v2._1,v1._2+v2._2,v1._3+v2._3)
      }
    ).sortBy(_._2,false).take(10).foreach(println)












  }
}
